package timers

import (
	"encoding/base64"
	"fmt"
	"gitee.com/mootok/mootok/src/constant/config"
	"gitee.com/mootok/mootok/src/external_service/censor"
	"gitee.com/mootok/mootok/src/external_service/transcoding"
	"gitee.com/mootok/mootok/src/models"
	"gitee.com/mootok/mootok/src/services/video"
	"gitee.com/mootok/mootok/src/storage/database"
	"gitee.com/mootok/mootok/src/utils/logging"
	log "github.com/sirupsen/logrus"
	"sync"
	"time"
)

func queryAuditTasks(duration time.Duration) {
	logger := logging.LogService("Timer.QueryAuditTasks")
	ticker := time.NewTicker(duration)
	quit := make(chan struct{})
	go func() {
		for {
			select {
			case <-ticker.C:
				logger.Info("Querying auditing videos...")
				var tasks []models.Video
				database.Client.
					Where("job_id != ?", "").
					Where("status = ?", video.StatusAuditing).
					Find(&tasks)
				if len(tasks) == 0 {
					logger.Info("No pending auditing videos.")
					continue
				}
				wg := sync.WaitGroup{}
				wg.Add(1)
				for i, elem := range tasks {
					taskLogger := logger.WithFields(log.Fields{
						"user_id":  elem.UserId,
						"video_id": elem.ID,
						"job_id":   elem.JobId,
					})
					status, result ,err := censor.Instance.QueryAuditVideoTaskProgress(elem.JobId)
					if err != nil {
						taskLogger.Warningf("progress query failed: %v", err)
						continue
					}
					if status == censor.StatusFinished {
						if result != censor.ResultPass {
							tasks[i].Status = video.StatusAuditFailed
							taskLogger.Warning("censor server returns non pass")
						} else {
							taskLogger.Info("censor passed")
							tasks[i].Status = video.StatusTranscoding
							tasks[i].JobId = ""
							go func() {
								// 转码
								// 等待数据库操作完毕
								taskLogger.Info("begin transcoding process")
								encodedUri := base64.URLEncoding.EncodeToString([]byte(config.EnvConfig.QiNiuKodoBucket + ":" + elem.FileName + ".m3u8"))

								resolutionOps := "multiResolution/"
								envBandWidthOps := "envBandWidth/"
								multiVbOps := "envBandWidth/"
								multiPrefixOps := "multiPrefix/"
								for index, policy := range *VideoService.GetTranscodePolicy() {
									resolutionOps = resolutionOps + fmt.Sprintf("%d:%d", policy.Resolution[0], policy.Resolution[1])
									envBandWidthOps = envBandWidthOps + fmt.Sprintf("%d", policy.Bitrate)
									multiVbOps = multiVbOps + fmt.Sprintf("%d", policy.Bitrate)
									multiPrefixOps = multiPrefixOps + base64.URLEncoding.EncodeToString([]byte(VideoService.GetTranscodeUrl(elem.FileName, policy)))
									if index != len(*VideoService.GetTranscodePolicy())-1 {
										resolutionOps = resolutionOps + ","
										envBandWidthOps = envBandWidthOps + ","
										multiVbOps = multiVbOps + ","
										multiPrefixOps = multiPrefixOps + ","
									}
								}
								persistentOps := fmt.Sprintf("adapt/m3u8/%s/%s/%s/%s/hlstime/10|saveas/%s;", resolutionOps, envBandWidthOps, multiVbOps, multiPrefixOps, encodedUri)
								jobId, err := transcoding.Instance.CreateTranscodingTask(elem.FileName, persistentOps)
								if err != nil {
									taskLogger.Errorf("starting pfop operation failed: %v", err)
									return
								} else {
									tasks[i].JobId = jobId
								}
								if err := database.Client.Select("job_id").Updates(tasks[i]).Error; err != nil {
									taskLogger.Errorf("failed to update tasks: %v", err)
									return
								}

							}()
						}
					} else if status == censor.StatusFailed {
						tasks[i].Status = video.StatusAuditFailed
						taskLogger.Warning("censor server returns failed")
					}
					if err := database.Client.Select("status").Updates(tasks[i]).Error; err != nil {
						taskLogger.Errorf("failed to update task: %v", err)
						continue
					}
				}
				// Update completed
				logger.Infof("processed %d tasks.", len(tasks))

			case <-quit:
				ticker.Stop()
				return
			}
		}
	}()
}